package test;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.lsda.job.*;
import org.lsda.util.*;


import java.util.*;

/**
 * This class is used to send the job request and ask for computation.
 * 
 * @author Chunqiu Zeng
 * 
 */
public class JobRequest implements MessageListener, ExceptionListener {

	private static Log log = LogFactory.getLog(JobRequest.class);

	/**
	 * This is used to denote the message queue for computing job
	 */
	private String queueName = Util.getBRQueName();

	/**
	 * url for the message broker
	 */
	private String url = Util.getBRUrl();

	/**
	 * user name for accessing the message broker
	 */
	private String userName = Util.getBRUser();
	/**
	 * password for accessing the message broker
	 */
	private String password = Util.getBRPasswd();

	/**
	 * whether to support the transaction
	 */
	private boolean transacted = true;

	/**
	 * used to control the request
	 */
	private boolean running = false;

	/**
	 * used to describe the information of job
	 */
	private Job job = null;
	
	/**
	 * used to store all the handlers for message
	 * */
	private ArrayList<ActionHandler> actionHandlers = new ArrayList<ActionHandler>();

	/**
	 * defualt constructor
	 * */
	public JobRequest() {
	}

	/**
	 * constructor
	 * */
	public JobRequest(Job job) throws Exception {
		this.job = job;
		if (!registerJob())
			throw new Exception("failed to register job " + job.getJobId());
	}

	/**
	 * get the queue name for the broker
	 * */
	public String getQueueName() {
		return queueName;
	}

	/**
	 * set the queue name for the broker
	 * */
	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public String getUrl() {
		return url;
	}

	public void setUrl(String url) {
		this.url = url;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public Job getJob() {
		return job;
	}
	
	public void addActionHandler(ActionHandler ah){
		actionHandlers.add(ah);
	}
	
	public void clearActionHandler(){
		actionHandlers.clear();
	}
	
	public void setJob(Job job) throws Exception {
		this.job = job;
		if (!registerJob())
			throw new Exception("failed to register job " + job.getJobId());
	}

	public synchronized void setRunning(boolean running) {
		this.running = running;
	}

	public synchronized boolean getRunning() {
		return running;
	}

	public boolean registerJob() {
		if (job == null)
			return false;
		boolean ret = false;
		String jobsql = "insert into job(jobid,jobname,jobdesc,status,rqstime)values(?,?,?,-2,now())";
		java.sql.Connection conn = null;
		try {
			conn = Util.getConnection();
			conn.setAutoCommit(false);
			java.sql.PreparedStatement pstmt = conn.prepareStatement(jobsql);
			pstmt.setInt(1, job.getJobId());
			pstmt.setString(2, job.getJobName());
			pstmt.setString(3, job.getJobDesc());
			pstmt.execute();
			conn.commit();
			pstmt.close();
			ret = true;
		} catch (Exception ex) {
			log.error(ex.getMessage());
		} finally {
			try {
				conn.close();
			} catch (Throwable igrone) {
			}
		}

		return ret;
	}

	public boolean updateJobStatus(int status) {
		if (job == null)
			return false;
		boolean ret = false;
		String jobsql = "update job set status=? where jobid=?";
		java.sql.Connection conn = null;
		try {
			conn = Util.getConnection();
			conn.setAutoCommit(false);
			java.sql.PreparedStatement pstmt = conn.prepareStatement(jobsql);
			pstmt.setInt(1, status);
			pstmt.setInt(2, job.getJobId());
			pstmt.execute();
			conn.commit();
			pstmt.close();
			ret = true;
		} catch (Exception ex) {
			log.error(ex.getMessage());
		} finally {
			try {
			} catch (Throwable ignore) {
			}
		}
		return ret;
	}

	public void doJob() {
		if (null == job)
			return;
		log.debug("update job("+job.getJobId()+") to request state") ;
		updateJobStatus(-1);
		setRunning(true);
		ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
				userName, password, url);
		Connection connection = null;
		try {
			connection = connectionFactory.createConnection();
			connection.start();
			Session session = connection.createSession(transacted,
					Session.AUTO_ACKNOWLEDGE);

			Destination destination = session.createQueue(queueName);
			MessageProducer msgProd = session.createProducer(destination);

			Destination tempDest = session.createTemporaryQueue();
			MessageConsumer msgCons = session.createConsumer(tempDest);

			// Thread.currentThread().getId()connection;
			ObjectMessage msg = session.createObjectMessage(job);
			msg.setJMSReplyTo(tempDest);
			msg.setJMSCorrelationID("" + job.getJobId());
			
			log.debug("send the job request to message broker");
			msgProd.send(msg);
			session.commit();
			int i=0;
			for (i = 0; i < job.getSize() && getRunning();) {
				Message message = msgCons.receive();
				if (message instanceof ObjectMessage) {
					i++;
					onMessage(message);
					session.commit();
				}
			}
			
			if(i==job.getSize()&&getRunning())
				updateJobStatus(0);

		} catch (Exception ex) {
			log.error(ex.getMessage());
		} finally {
			try {
				connection.close();
			} catch (Throwable ignore) {
			}
		}
	}

	public void doAct(Action act){
		ActionHandler ah = null;
		for(int i=0;i<actionHandlers.size();i++)
		{
			ah = actionHandlers.get(i);
//			ah.handle(act);
		}
		log.info("reply:" + act.getPrgName() + " " + act.getParam() + " "
				+ act.getStatus()+""+act.getInfo());
	}
	
	@Override
	public void onMessage(Message arg0) {
		// TODO Auto-generated method stub
		ObjectMessage om = (ObjectMessage) arg0;
		try {
			Action act = (Action) om.getObject();
			if (act.getStatus() != 0) {
				setRunning(false);
				updateJobStatus(act.getStatus());
			}
			job.setAction(act);
			
			doAct(act);
			
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			log.error(e.getMessage());
		}
	}

	@Override
	public void onException(JMSException arg0) {
		// TODO Auto-generated method stub
		log.error(arg0.getMessage());
		setRunning(false);
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		Job job = new Job();
		job.setJobName("reprocess");
		job.setJobDesc("this is used to reprocess data");
		Action act1 = new Action();
		act1.setSubJobId(0);
		act1.setPrgName("ls");
		act1.setParam(" -l ");
		job.addAction(act1);
		Action act2 = new Action();
		act2.setSubJobId(1);
		act2.setPrgName("echo ");
		act2.setParam(" \"hello world\"");
		job.addAction(act2);
		JobRequest jr = new JobRequest(job);
		jr.doJob();
	}

}
